/*
   Copyright (c) 2016 VMware, Inc. All Rights Reserved.
   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

package job

import (
	"github.com/vmware/harbor/src/common/dao"
	"github.com/vmware/harbor/src/common/models"
	"github.com/vmware/harbor/src/common/utils/log"
	"github.com/vmware/harbor/src/jobservice/config"
)

//定义workerPool池子
type workerPool struct {
	workerChan chan *Worker
	workerList []*Worker
}

// WorkerPool is a set of workers each worker is associate to a statemachine for handling jobs.
// it consists of a channel for free workers and a list to all workers
var WorkerPool *workerPool

// StopJobs accepts a list of jobs and will try to stop them if any of them is being executed by the worker.
//停止给定任务ID的调度任务
func (wp *workerPool) StopJobs(jobs []int64) {
	log.Debugf("Works working on jobs: %v will be stopped", jobs)
	for _, id := range jobs {
		for _, w := range wp.workerList {
			if w.SM.JobID == id {
				log.Debugf("found a worker whose job ID is %d, will try to stop it", id)
				//在状态机里停止调度任务
				w.SM.Stop(id)
			}
		}
	}
}

// Worker consists of a channel for job from which worker gets the next job to handle, and a pointer to a statemachine,
// the actual work to handle the job is done via state machine.
//定义worker结构体
type Worker struct {
	//workerID
	ID int
	//镜像复制任务队列
	RepJobs chan int64
	//状态机
	SM *SM
	//是否停止worker
	quit chan bool
}

// Start is a loop worker gets id from its channel and handle it.
//启动workers
func (w *Worker) Start() {
	go func() {
		for {
			//把自己注册到WorkerPool.workerChan channel中
			WorkerPool.workerChan <- w
			select {
			//如果w.RepJobs镜像复制队列当中存在任务,就获取然后交由w.handleRepJob(jobID) 处理
			case jobID := <-w.RepJobs:
				log.Debugf("worker: %d, will handle job: %d", w.ID, jobID)
				w.handleRepJob(jobID)
			case q := <-w.quit:
				if q {
					log.Debugf("worker: %d, will stop.", w.ID)
					return
				}
			}
		}
	}()
}

// Stop ...
//停止worker
func (w *Worker) Stop() {
	go func() {
		w.quit <- true
	}()
}

//处理worker接收到的任务
func (w *Worker) handleRepJob(id int64) {
	//重置任务状态
	err := w.SM.Reset(id)
	if err != nil {
		log.Errorf("Worker %d, failed to re-initialize statemachine for job: %d, error: %v", w.ID, id, err)
		err2 := dao.UpdateRepJobStatus(id, models.JobError)
		if err2 != nil {
			log.Errorf("Failed to update job status to ERROR, job: %d, error:%v", id, err2)
		}
		return
	}
	//Enabled == 0 关闭同步任务
	if w.SM.Parms.Enabled == 0 {
		log.Debugf("The policy of job:%d is disabled, will cancel the job", id)
		_ = dao.UpdateRepJobStatus(id, models.JobCanceled)
		w.SM.Logger.Info("The job has been canceled")
	} else {
		//启动状态机
		w.SM.Start(models.JobRunning)
	}
}

// NewWorker returns a pointer to new instance of worker
//创建worker
func NewWorker(id int) *Worker {
	w := &Worker{
		//workerId
		ID: id,
		//镜像复制队列
		RepJobs: make(chan int64),
		//是否退出
		quit: make(chan bool),
		//状态机
		SM: &SM{},
	}
	//初始化状态机
	w.SM.Init()
	return w
}

// InitWorkerPool create workers according to configuration.
//初始化工人池
func InitWorkerPool() {
	WorkerPool = &workerPool{
		workerChan: make(chan *Worker, config.MaxJobWorkers()),
		workerList: make([]*Worker, 0, config.MaxJobWorkers()),
	}
	for i := 0; i < config.MaxJobWorkers(); i++ {
		//创建worker
		worker := NewWorker(i)
		WorkerPool.workerList = append(WorkerPool.workerList, worker)
		//启动worker
		worker.Start()
		log.Debugf("worker %d started", worker.ID)
	}
}

// Dispatch will listen to the jobQueue of job service and try to pick a free worker from the worker pool and assign the job to it.
//循环获取jobQueue的任务,如果获取到任务就从WorkerPool.workerChan池中取出一个工人(worker) 进行处理
func Dispatch() {
	for {
		select {
		case job := <-jobQueue:
			go func(jobID int64) {
				log.Debugf("Trying to dispatch job: %d", jobID)
				worker := <-WorkerPool.workerChan
				worker.RepJobs <- jobID
			}(job)
		}
	}
}
